package de.mmis.core.publishsubscribe;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Modifier;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.log4j.Logger;

import de.mmis.core.base.MethodDescriptor;
import de.mmis.core.base.ObjectDescriptor;
import de.mmis.core.base.Pair;
import de.mmis.core.base.abstracttree.InnerNode;
import de.mmis.core.base.abstracttree.Tree;
import de.mmis.core.base.abstracttree.Tree.Type;
import de.mmis.core.base.abstracttree.TreeParserException;
import de.mmis.core.base.abstracttree.serializer.DeserializableException;
import de.mmis.core.base.event.Event;
import de.mmis.core.base.event.Observable;
import de.mmis.core.base.event.Observer;
import de.mmis.core.config.Config;
import de.mmis.core.publishsubscribe.FilterHelper.ExtractResult;
import de.mmis.core.publishsubscribe.FilterManager.FilterEntry;
import de.mmis.core.publishsubscribe.filter.Filter;
import de.mmis.core.sexpression.SExpressionParser;
import de.mmis.core.sexpression.SExpressionWriter;

public class PublishSubscribeImpl implements PublishSubscribe {

	private final static Logger LOGGER = Logger
			.getLogger(PublishSubscribeImpl.class);

	private class PublishSubscribeThread extends Thread {

		public PublishSubscribeThread() throws PublishSubscribeException {
			super(PublishSubscribeThread.class.getName());
			setDaemon(true);
			readHello();
		}

		private void readHello() throws PublishSubscribeException {
			LOGGER.trace("waiting for hello message");

			try {
				Tree h = SExpressionParser.parse(input, false);
				if (h.getType() != Type.INNER) {
					throw new TreeParserException(
							"hello message is not a struct");
				}
				InnerNode hello = (InnerNode) h;
				remoteBindAddress = InetAddress.getByName(hello.getSubTree(1)
						.getAsString());
				LOGGER.trace("received hello, bind address set to "
						+ remoteBindAddress);
			} catch (TreeParserException e) {
				throw new PublishSubscribeException(
						"Error receiving hello message", e);
			} catch (IOException e) {
				throw new PublishSubscribeException(
						"Error receiving hello message", e);
			} catch (DeserializableException e) {
				throw new PublishSubscribeException(
						"Error receiving hello message", e);
			}

			synchronized (ready) {
				ready.set(true);
				ready.notifyAll();
			}
		}

		@Override
		public void run() {
			while (running) {
				while (!clientSocket.isClosed()) {
					Tree expr = null;
					try {
						expr = SExpressionParser.parse(input, false);
					} catch (TreeParserException e) {
						if (running && !e.unexpectedEndOfStreamOccurred())
							LOGGER.error("Can't parse input from server", e);
						break;
					} catch (IOException e) {
						if (running)
							LOGGER.error("Can't read input from server", e);
						break;
					}

					if (expr == null)
						continue;

					Message message;
					try {
						message = expr.getAs(Message.class);
					} catch (DeserializableException e) {
						LOGGER.error("Can't deserialize message: " + expr, e);
						continue;
					}

					parseMessage(message);
				}

				synchronized (ready) {
					ready.set(false);
				}

				if (!running)
					break;

				LOGGER.info("Setting up new server");

				// try to create new server and connect to it
				try {
					setUpServer();
					// receive hello message
					readHello();
				} catch (PublishSubscribeException e) {
					LOGGER.error("Could not set up new Server", e);
					running = false;
					break;
				}

				// send subscription for each filter
				for (FilterEntry<?, Class<? extends Event>> entry : inputFilters
						.getAllEntries())
					if (!sendSubscribeMessage(entry.getFilter(),
							entry.getClazz(), entry.getID())) {
						LOGGER.fatal("New socket was unable to send - give up");
						running = false;
					}
			} // while (running)
		}
	}

	private static PublishSubscribeImpl instance;
	private final static Object instanceMonitor = new Object();

	public static PublishSubscribeImpl getInstance()
			throws PublishSubscribeException {
		synchronized (instanceMonitor) {
			if (instance == null)
				instance = new PublishSubscribeImpl();
			return instance;
		}
	}

	private volatile boolean running = false;
	private AtomicBoolean ready = new AtomicBoolean(false);

	private InetAddress remoteBindAddress = null;
	private Socket clientSocket;
	private BufferedInputStream input;
	private BufferedOutputStream output;

	private final HashMap<Observable<?>, Observer<?>> observerMap = new HashMap<Observable<?>, Observer<?>>();
	private final HashMap<Observable<?>, String> producerIDs = new HashMap<Observable<?>, String>();

	/**
	 * {@link Filter}s local observers have subscribed. Must be checked against
	 * incoming {@link Event}s
	 */
	@SuppressWarnings("rawtypes")
	private final FilterManager<Pair<Observer, Observer<PlainEvent>>, Class<? extends Event>> inputFilters = new FilterManager<Pair<Observer, Observer<PlainEvent>>, Class<? extends Event>>();

	private final HashMap<Filter, Integer> worldstateFilter = new HashMap<Filter, Integer>();

	/** only negative values to distinguish from {@link #inputFilters} IDs */
	private final AtomicInteger lastWorldstateFilterID = new AtomicInteger(-1);

	private final HashMap<String, Map<String, Tree>> worldstate = new HashMap<String, Map<String, Tree>>();

	// /**
	// * Stores the filter of the {@link HostManager}, the "client" being the id
	// * of the HostManager
	// */
	// private final FilterManager<Integer, Object> outputFilters = new
	// FilterManager<Integer, Object>();

	private PublishSubscribeImpl() throws PublishSubscribeException {
		// singleton, hide constructor
		start();
	}

	public void start() throws PublishSubscribeException {
		if (running)
			return;

		setUpServer();
		PublishSubscribeThread thread = new PublishSubscribeThread();
		running = true;
		thread.start();
	}

	private void setUpServer() throws PublishSubscribeException {
		try {
			createSocket();
			if (clientSocket == null) {
				HostManager manager = HostManager
						.getInstance(remoteBindAddress);
				PublishSubscribeException serverException = null;
				// there is no server running - create one
				try {
					manager.init();
				} catch (PublishSubscribeException e) {
					// unable to create servers, mostly because
					// we were too slow, so there should now be another server

					// save exception
					serverException = e;
				} finally {
					// wait so the server should be ready to accept connections
					try {
						Thread.sleep(250);
					} catch (InterruptedException e) {
						// ignore
					}

					// connect to local server
					createSocket();

					if (clientSocket == null) {
						throw new PublishSubscribeException(
								"Unable to connect to local server",
								serverException);
					}
				}
			}
		} catch (PublishSubscribeException e) {
			running = false;
			throw e;
		}
	}

	private void createSocket() {
		try {
			clientSocket = new Socket((String) null, Config.PUBSUB_PORT);
			input = new BufferedInputStream(clientSocket.getInputStream());
			output = new BufferedOutputStream(clientSocket.getOutputStream());
		} catch (UnknownHostException e) {
			// should not happen
			LOGGER.error("", e);
		} catch (IOException e) {
			clientSocket = null;
			input = null;
			output = null;
		}
	}

	private void parseMessage(Message message) {
		LOGGER.trace("parsing message: " + message);

		switch (message.getType()) {
		case EVENT:
			processEvent(message);
			break;

		case SUBSCRIBE:
			// // client == msg.id
			// outputFilters.addFilter(message.getClazz(), message.getFilter(),
			// message.getId(), null);
			// break;
		case UNSUBSCRIBE:
			// // client == msg.id
			// for (Integer id : message.getIds())
			// outputFilters.removeClient(id);
			// break;
			LOGGER.warn("remote un/subscribe not supported");
			break;

		case STATE_ANSWER:
			synchronized (worldstate) {
				Map<String, Tree> state = worldstate.get(message.getDeviceID());
				if (state == null || !state.containsKey(message.getKey()))
					break;
				state.put(message.getKey(), message.getValue());
			}
			break;

		case STATE_REQUEST:
			String id = message.getDeviceID();
			if (!producerIDs.containsValue(id))
				// only answer when we manage the producer
				break;

			Tree value;
			synchronized (worldstate) {
				Map<String, Tree> state = worldstate.get(id);
				if (state == null)
					break;
				value = state.get(message.getKey());
				if (value == null)
					break;
			}
			Message m = Message.createStateAnswer(id, message.getKey(), value,
					message.getDestIP());
			sendMessage(m);
			break;
		}
	}

	/**
	 * updates the worldstate and delivers the event to other observers
	 * 
	 * @param message
	 */
	@SuppressWarnings("unchecked")
	private void processEvent(Message message) {
		synchronized (worldstate) {
			PlainEvent event = message.getEvent();

			// update worldstate
			Map<String, Tree> eventData = event.eventData;
			Tree sender = eventData.get(":sender");
			if (sender != null) {
				Map<String, Tree> state;
				try {
					String id = sender.getAsString();
					state = worldstate.get(id);
					if (state != null)
						for (String key : eventData.keySet())
							if (state.containsKey(key)) {
								LOGGER.trace("Updating key " + key
										+ " of device " + id
										+ " in the worldstate");
								state.put(key, eventData.get(key));
							}
				} catch (DeserializableException e) {
					LOGGER.error(sender + " is not a string", e);
				}
			}

			if (!message.getDeliver())
				return;

			// check filters if any client is interested in the event
			for (@SuppressWarnings("rawtypes")
			LinkedList<FilterEntry<Pair<Observer, Observer<PlainEvent>>, Class<? extends Event>>> l : inputFilters
					.getForEvent(event, worldstate).values())
				for (@SuppressWarnings("rawtypes")
				FilterEntry<Pair<Observer, Observer<PlainEvent>>, Class<? extends Event>> entry : l) {
					@SuppressWarnings("rawtypes")
					Pair<Observer, Observer<PlainEvent>> pair = entry
							.getClient();

					if (pair.getFirst() != null) {
						Event e = EventHelper.plain2event(event,
								entry.getInfo());
						pair.getFirst().notify(null, e);
					}

					if (pair.getSecond() != null)
						pair.getSecond().notify(null, event);
				}
		}
	}

	/**
	 * Sends message to local server
	 * 
	 * @return true if message was sent, false otherwise
	 */
	private boolean sendMessage(Message message) {
		Tree e = Tree.fromObject(message);
		LOGGER.trace("Sending message to local server: " + message);

		synchronized (ready) {
			while (!ready.get())
				try {
					ready.wait();
				} catch (InterruptedException e1) {
					// should not happen
					LOGGER.error("", e1);
				}
		}

		byte[] b = SExpressionWriter.writeSExpression(e);
		try {
			output.write(b);
			output.flush();
			return true;
		} catch (IOException ex) {
			LOGGER.error("Could not send message", ex);
			return false;
		}
	}

	@Override
	public <T extends Event> void addObservable(Observable<T> producer,
			final String id) throws PublishSubscribeException {
		synchronized (observerMap) {
			if (observerMap.containsKey(producer))
				return;

			LOGGER.debug("Adding Observable " + id + ": " + producer);

			producerIDs.put(producer, id);

			initWorldstate(id, producer);

			Observer<T> o = new Observer<T>() {
				@Override
				public void notify(Observable<? extends T> sender, T event) {
					PlainEvent e = EventHelper.event2plain(event, id);
					Message m = Message.createEvent(e);
					processEvent(m);
					sendMessage(m);
				}
			};

			producer.addObserver(o);
			observerMap.put(producer, o);
		}

		return;
	}

	private void initWorldstate(String id, Observable<?> producer) {
		synchronized (worldstate) {
			Map<String, Tree> state = worldstate.get(id);
			if (state == null) {
				state = new HashMap<String, Tree>();
				worldstate.put(id, state);
			}

			ObjectDescriptor descr = new ObjectDescriptor(producer, false);
			for (MethodDescriptor m : descr.getGetterMethodDescriptors()) {
				String key = m.getGetterName();

				Object result;
				try {
					result = m.invoke();
				} catch (IllegalArgumentException e) {
					LOGGER.error(
							"Could not init worldstate: invoking getter method "
									+ m + " failed", e);
					continue;
				} catch (IllegalAccessException e) {
					LOGGER.error(
							"Could not init worldstate: invoking getter method "
									+ m + " failed", e);
					continue;
				} catch (InvocationTargetException e) {
					LOGGER.error(
							"Could not init worldstate: invoking getter method "
									+ m + " failed", e);
					continue;
				}
				Tree value = Tree.fromObject(result, false);

				state.put(key, value);
			}

			// fire event with the current state, so if the physical device
			// was changed while the producer was disconnected, consumers get
			// the new state
			HashSet<String> classes = new HashSet<String>();
			classes.add(Event.class.getName());
			PlainEvent event = new PlainEvent(Event.class.getName(), classes,
					state, null);

			// don't deliver this event to consumers, just update the worldstate
			Message m = Message.createEvent(event, false);
			sendMessage(m);
		}
	}

	@Override
	public <T extends Event> void removeObservable(Observable<T> producer)
			throws PublishSubscribeException {
		LOGGER.debug("removing observable " + producer);

		synchronized (observerMap) {
			@SuppressWarnings("unchecked")
			Observer<T> o = (Observer<T>) observerMap.remove(producer);
			producerIDs.remove(producer);

			if (o == null)
				return;

			producer.removeObserver(o);
		}
	}

	@Override
	public <T extends Event> Integer subscribe(Class<T> clazz, Filter filter,
			Observer<? super T> consumer) throws PublishSubscribeException {
		return subscribe(clazz, filter, consumer, null);
	}

	@Override
	public <T extends Event> Integer subscribe(Class<T> clazz, Filter filter,
			Observer<? super T> consumer, Observer<PlainEvent> plainConsumer)
			throws PublishSubscribeException {
		if (consumer == null && plainConsumer == null)
			return null;

		if (Modifier.isAbstract(clazz.getModifiers())
				&& !Modifier.isInterface(clazz.getModifiers()))
			throw new PublishSubscribeException(
					"Can't subscribe for abstract classes");

		// handle original filter
		@SuppressWarnings("rawtypes")
		Pair<Observer, Observer<PlainEvent>> pair = new Pair<Observer, Observer<PlainEvent>>(
				consumer, plainConsumer);
		Integer id = inputFilters.addFilter(clazz.getName(), filter, pair,
				clazz);

		LOGGER.debug("new subscription, class = " + clazz + ", filter = "
				+ filter + ", consumer = " + consumer + ", plainConsumer = "
				+ plainConsumer + ", id = " + id);

		sendSubscribeMessage(filter, clazz.getName(), id);

		synchronized (worldstateFilter) {

			// handle extracted worldstate filters
			ExtractResult result = FilterHelper.extractWorldstateFilter(filter);
			for (Entry<Filter, Pair<String, String>> entry : result.worldstateFilter
					.entrySet()) {
				Filter f = entry.getKey();
				if (worldstateFilter.containsKey(f))
					// we already get these events
					continue;

				// remember: this filterID is negative, thus decrement here
				Integer ID = new Integer(
						lastWorldstateFilterID.decrementAndGet());
				worldstateFilter.put(f, ID);

				// if the worldstate does not contain this key, add "null"
				// if we receive any event, we check in the worldstate-Map if we
				// are interested in the value

				String deviceID = entry.getValue().getFirst();
				String key = entry.getValue().getSecond();

				synchronized (worldstate) {
					Map<String, Tree> ed = worldstate.get(deviceID);
					if (ed == null) {
						LOGGER.trace("Adding device " + deviceID
								+ " to the worldstate");
						ed = new HashMap<String, Tree>();
						worldstate.put(deviceID, ed);

						// request the current state
						sendMessage(Message.createStateRequest(deviceID, key,
								null));
					}

					if (!ed.containsKey(key)) {
						LOGGER.trace("Adding key " + key + " of device "
								+ deviceID + " to the worldstate");
						ed.put(key, null);
					}
				}
				// subscribe for this event
				sendSubscribeMessage(f, Event.class.getName(), ID);
			}
		}

		return id;
	}

	/**
	 * Sends a subscription message to the local server corresponding to the
	 * entry
	 * 
	 * @return true iff the message could be sent.
	 */
	private boolean sendSubscribeMessage(Filter filter, String clazz, Integer id) {
		Message m = Message.createSubscribe(filter, clazz, id);
		return sendMessage(m);
	}

	/**
	 * @return a {@link List} of all IDs of removed {@link #worldstateFilter},
	 *         used to send unsubscribe messages
	 */
	private LinkedList<Integer> rebuildWorldstateFilter() {
		LOGGER.trace("rebuilding worldstateFilter");

		LinkedList<Integer> result = new LinkedList<Integer>();

		// reconstruct all worldstateFilters and check if we don't need some
		HashMap<Filter, Pair<String, String>> filters = new HashMap<Filter, Pair<String, String>>();

		for (FilterEntry<?, Class<? extends Event>> filterEntry : inputFilters
				.getAllEntries()) {
			HashMap<Filter, Pair<String, String>> fs;
			fs = FilterHelper.extractWorldstateFilter(filterEntry.getFilter()).worldstateFilter;
			filters.putAll(fs);
		}

		synchronized (worldstateFilter) {
			Iterator<Entry<Filter, Integer>> it = worldstateFilter.entrySet()
					.iterator();
			while (it.hasNext()) {
				Entry<Filter, Integer> wsfEntry = it.next();
				Filter f = wsfEntry.getKey();
				if (filters.containsKey(f))
					continue;

				result.add(wsfEntry.getValue());
				it.remove();
			}
		}

		// remove entries from worldstate
		synchronized (worldstate) {
			Collection<Pair<String, String>> values = filters.values();

			Iterator<Entry<String, Map<String, Tree>>> it = worldstate
					.entrySet().iterator();
			while (it.hasNext()) {
				// for every ID - KEY entry in the worldstate,
				// check if this pair is in the values collection,
				// i. e. if it's still needed
				Entry<String, Map<String, Tree>> entry = it.next();
				String id = entry.getKey();
				Map<String, Tree> state = entry.getValue();

				Iterator<String> itState = state.keySet().iterator();
				while (itState.hasNext()) {
					String key = itState.next();
					Pair<String, String> pair = new Pair<String, String>(id,
							key);
					if (values.contains(pair))
						// this key is still needed
						continue;

					// this key is not needed any longer
					LOGGER.trace("removing key " + key + " of device " + id
							+ " from the worldstate");
					itState.remove();
				}

				if (state.isEmpty()) {
					// the hole device is not needed any longer
					LOGGER.trace("remove device " + id + " from the worldstate");
					it.remove();
				}
			}
		}

		return result;
	}

	@Override
	public void unsubscribe(Integer id) {
		if (id == null)
			return;

		LOGGER.debug("Unsubscription of id " + id);

		FilterEntry<?, Class<? extends Event>> entry = inputFilters
				.removeFilter(id);
		if (entry == null) {
			LOGGER.warn("unsubscribe: id " + id + " not found");
			return;
		}

		LinkedList<Integer> ids = new LinkedList<Integer>();
		ids.add(id);

		// find out if any worldstate filters can be unsubscribed, too

		ExtractResult result = FilterHelper.extractWorldstateFilter(entry
				.getFilter());
		// cheap check
		if (!result.worldstateFilter.isEmpty())
			ids.addAll(rebuildWorldstateFilter());

		Message m = Message.createUnsubscribe(ids);
		sendMessage(m);
	}

	// @Override
	// public void unsubscribeAll(Observer<?> consumer) {
	// LOGGER.debug("Unsubscribe all of " + consumer);
	//
	// LinkedList<Integer> ids = new LinkedList<Integer>();
	//
	// @SuppressWarnings("rawtypes")
	// LinkedList<FilterEntry<Observer, Class<? extends Event>>> entries =
	// inputFilters
	// .removeClient(consumer);
	//
	// boolean rebuild = false;
	//
	// for (@SuppressWarnings("rawtypes")
	// FilterEntry<Observer, Class<? extends Event>> entry : entries) {
	// ids.add(entry.getID());
	// if (!rebuild) {
	// ExtractResult result = FilterHelper
	// .extractWorldstateFilter(entry.getFilter());
	// // cheap check
	// if (!result.worldstateFilter.isEmpty())
	// rebuild = true;
	// }
	// }
	//
	// if (rebuild)
	// ids.addAll(rebuildWorldstateFilter());
	//
	// Message m = Message.createUnsubscribe(ids);
	// sendMessage(m);
	// }

}
